/*!
 * \file pool_core.hpp
 * \brief 线程池核心实现
 *
 * \details 此文件包含线程池的核心类：pool<Task, SchedulingPolicy>。
 *          
 *          线程池是在同一进程内进行异步和并行处理的机制。pool类提供了
 *          一种方便的方式来分发异步任务作为函数对象。这些任务的调度
 *          可以通过使用自定义的调度器轻松控制。
 *          
 *          核心特性：
 *          - 高度模板化的设计，支持多种策略组合
 *          - 线程安全的任务调度和执行
 *          - 动态线程池大小管理
 *          - 多种调度策略支持（FIFO、LIFO、优先级等）
 *          - 优雅的关闭和资源管理
 *          - 任务等待和同步机制
 *          
 *          适用场景：
 *          - 高频交易系统的并行计算
 *          - 大规模数据处理的任务分发
 *          - 网络服务的请求处理
 *          - I/O密集型操作的异步执行
 *
 * Copyright (c) 2005-2007 Philipp Henkel
 *
 * Use, modification, and distribution are  subject to the
 * Boost Software License, Version 1.0. (See accompanying  file
 * LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
 *
 * http://threadpool.sourceforge.net
 *
 */


#ifndef THREADPOOL_POOL_CORE_HPP_INCLUDED
#define THREADPOOL_POOL_CORE_HPP_INCLUDED




#include "locking_ptr.hpp"
#include "worker_thread.hpp"

#include "../task_adaptors.hpp"

#include <boost/thread.hpp>
#include <boost/thread/exceptions.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/thread/condition.hpp>
#include <boost/smart_ptr.hpp>
#include <boost/bind.hpp>
#include <boost/static_assert.hpp>
#include <boost/type_traits.hpp>

#include <vector>


/// threadpool命名空间包含线程池和相关的实用工具类
namespace boost { namespace threadpool { namespace detail 
{

  /*!
   * \brief 线程池核心实现类
   *
   * \details 线程池是在同一进程内进行异步和并行处理的机制。
   *          pool_core类提供了一种方便的方式来分发异步任务作为函数对象。
   *          这些任务的调度可以通过使用自定义的调度器轻松控制。
   *          任务不得抛出异常。
   *
   *          pool_core是DefaultConstructible和NonCopyable的。
   *          
   *          核心功能包括：
   *          - 任务调度和执行管理
   *          - 工作线程生命周期管理  
   *          - 动态线程池大小调整
   *          - 任务队列管理和优化
   *          - 线程安全的状态监控
   *          - 优雅关闭和资源清理
   *
   * \tparam Task 实现operator 'void operator() (void) const'的函数对象。
   *              pool调用此操作符来执行任务。异常将被忽略。
   * \tparam SchedulingPolicy 决定任务如何调度的任务容器。
   *                         保证此容器一次只能被一个线程访问。调度器不应抛出异常。
   * \tparam SizePolicy 线程池大小策略，控制线程数量的动态调整
   * \tparam SizePolicyController 大小策略控制器
   * \tparam ShutdownPolicy 关闭策略，定义线程池的关闭行为
   *
   * \warning pool_core类是线程安全的
   * 
   * \see 任务：task_func, prio_task_func
   * \see 调度策略：fifo_scheduler, lifo_scheduler, prio_scheduler
   */ 
  template <
    typename Task, 

    template <typename> class SchedulingPolicy,
    template <typename> class SizePolicy,
    template <typename> class SizePolicyController,
    template <typename> class ShutdownPolicy
  > 
  class pool_core
  : public enable_shared_from_this< pool_core<Task, SchedulingPolicy, SizePolicy, SizePolicyController, ShutdownPolicy > > 
  , private noncopyable
  {

  public: // 类型定义
    typedef Task task_type;                                 //!< 指示任务的类型
    typedef SchedulingPolicy<task_type> scheduler_type;     //!< 指示调度器的类型
    typedef pool_core<Task, 
                      SchedulingPolicy, 
                      SizePolicy,
                      SizePolicyController,
                      ShutdownPolicy > pool_type;           //!< 指示线程池的类型
    typedef SizePolicy<pool_type> size_policy_type;         //!< 指示大小策略的类型
    //typedef typename size_policy_type::size_controller size_controller_type;

    typedef SizePolicyController<pool_type> size_controller_type;

//    typedef SizePolicy<pool_type>::size_controller size_controller_type;
    typedef ShutdownPolicy<pool_type> shutdown_policy_type;//!< 指示关闭策略的类型

    typedef worker_thread<pool_type> worker_type;

    // 任务必须是无参数函数
    BOOST_STATIC_ASSERT(function_traits<task_type()>::arity == 0);

    // 任务函数的返回类型必须是void
    BOOST_STATIC_ASSERT(is_void<typename result_of<task_type()>::type >::value);


  private:  // 友元类
    friend class worker_thread<pool_type>;

#if defined(__SUNPRO_CC) && (__SUNPRO_CC <= 0x580)  // Tested with CC: Sun C++ 5.8 Patch 121018-08 2006/12/06
   friend class SizePolicy;
   friend class ShutdownPolicy;
#else
   friend class SizePolicy<pool_type>;
   friend class ShutdownPolicy<pool_type>;
#endif

  private: // 以下成员可能被多个线程同时访问：
    volatile size_t m_worker_count;	        //!< 当前工作线程数量
    volatile size_t m_target_worker_count;	//!< 目标工作线程数量
    volatile size_t m_active_worker_count;  //!< 活跃工作线程数量
      


  private: // 以下成员一次只能被一个线程访问：
    scheduler_type  m_scheduler;                                //!< 任务调度器
    scoped_ptr<size_policy_type> m_size_policy;                 //!< 大小策略（永不为null）
    
    bool  m_terminate_all_workers;								//!< 指示是否触发了所有工作线程的终止
    std::vector<shared_ptr<worker_type> > m_terminated_workers; //!< 已终止但尚未完全析构的工作线程列表
    
  private: // 以下成员实现了线程安全：
    mutable recursive_mutex  m_monitor;                         //!< 线程同步互斥锁
    mutable condition m_worker_idle_or_terminated_event;	    //!< 工作线程空闲或已终止事件
    mutable condition m_task_or_terminate_workers_event;        //!< 任务可用或应减少工作线程总数事件

  public:
    /*!
     * \brief 构造函数
     * 
     * \details 初始化线程池核心，设置初始状态：
     *          - 所有线程计数器为0
     *          - 未启动终止标志
     *          - 创建大小策略实例
     *          - 清空任务调度器
     */
    pool_core()
      : m_worker_count(0) 
      , m_target_worker_count(0)
      , m_active_worker_count(0)
      , m_terminate_all_workers(false)
    {
      pool_type volatile & self_ref = *this;
      m_size_policy.reset(new size_policy_type(self_ref));

      m_scheduler.clear();
    }


    /*!
     * \brief 析构函数
     * 
     * \details 清理线程池资源。
     *          注意：析构前应调用shutdown()确保优雅关闭。
     */
    ~pool_core()
    {
    }

    /*!
     * \brief 获取管理池中线程数量的大小控制器
     * \return 大小控制器实例
     * 
     * \details 大小控制器提供动态调整线程池大小的接口，
     *          可以根据工作负载自动或手动调整线程数量。
     * 
     * \see SizePolicy
     */
    size_controller_type size_controller()
    {
      return size_controller_type(*m_size_policy, this->shared_from_this());
    }

    /*!
     * \brief 获取池中的线程数量
     * \return 线程数量
     * 
     * \details 返回当前线程池中的总线程数，包括活跃和空闲的线程。
     *          这是一个原子操作，可以安全地从任何线程调用。
     */
    size_t size()	const volatile
    {
      return m_worker_count;
    }

// TODO 只调用一次
    /*!
     * \brief 关闭线程池
     * 
     * \details 根据关闭策略优雅地关闭线程池。
     *          这会触发所有工作线程的终止过程。
     */
    void shutdown()
    {
      ShutdownPolicy<pool_type>::shutdown(*this);
    }

    /*!
     * \brief 调度任务以供异步执行
     * \param task 任务函数对象，不应抛出异常
     * \return 如果任务能够被调度则返回true，否则返回false
     * 
     * \details 将任务添加到调度器队列中以供工作线程执行。
     *          任务只会被执行一次。调度成功后会通知等待的工作线程。
     *          
     *          线程安全：此函数可以从多个线程同时调用。
     */  
    bool schedule(task_type const & task) volatile
    {	
      locking_ptr<pool_type, recursive_mutex> lockedThis(*this, m_monitor); 
      
      if(lockedThis->m_scheduler.push(task))
      {
        lockedThis->m_task_or_terminate_workers_event.notify_one();
        return true;
      }
      else
      {
        return false;
      }
    }	


    /*!
     * \brief 返回当前正在执行的任务数量
     * \return 活跃任务的数量
     * 
     * \details 获取当前正在工作线程中执行的任务数量。
     *          这不包括队列中等待执行的任务。
     */  
    size_t active() const volatile
    {
      return m_active_worker_count;
    }


    /*!
     * \brief 返回准备执行的任务数量
     * \return 待处理任务的数量
     * 
     * \details 获取调度器队列中等待执行的任务数量。
     *          这不包括当前正在执行的任务。
     */  
    size_t pending() const volatile
    {
      locking_ptr<const pool_type, recursive_mutex> lockedThis(*this, m_monitor);
      return lockedThis->m_scheduler.size();
    }


    /*!
     * \brief 从池的调度器中移除所有待处理任务
     * 
     * \details 清空调度器队列，所有等待执行的任务将被丢弃。
     *          正在执行的任务不会被影响。
     */  
    void clear() volatile
    { 
      locking_ptr<pool_type, recursive_mutex> lockedThis(*this, m_monitor);
      lockedThis->m_scheduler.clear();
    }    


    /*!
     * \brief 指示没有待处理任务
     * \return 如果没有准备执行的任务则返回true
     * 
     * \details 检查调度器队列是否为空。
     *          此函数比检查'pending() == 0'更高效。
     */   
    bool empty() const volatile
    {
      locking_ptr<const pool_type, recursive_mutex> lockedThis(*this, m_monitor);
      return lockedThis->m_scheduler.empty();
    }	


    /*!
     * \brief 阻塞当前执行线程，直到所有活跃和待处理任务的总和等于或小于给定阈值
     * \param task_threshold 池和调度器中任务的最大数量
     * 
     * \details 等待线程池达到指定的负载水平。默认阈值为0，
     *          表示等待所有任务完成。这对于同步和确保任务完成很有用。
     */     
    void wait(size_t const task_threshold = 0) const volatile
    {
      const pool_type* self = const_cast<const pool_type*>(this);
      recursive_mutex::scoped_lock lock(self->m_monitor);

      if(0 == task_threshold)
      {
        while(0 != self->m_active_worker_count || !self->m_scheduler.empty())
        { 
          self->m_worker_idle_or_terminated_event.wait(lock);
        }
      }
      else
      {
        while(task_threshold < self->m_active_worker_count + self->m_scheduler.size())
        { 
          self->m_worker_idle_or_terminated_event.wait(lock);
        }
      }
    }	

    /*!
     * \brief 阻塞当前执行线程直到满足时间戳或所有活跃和待处理任务的总和等于或小于给定阈值
     * \param timestamp 函数最晚返回的时间
     * \param task_threshold 池和调度器中任务的最大数量
     * \return 如果任务总和等于或小于阈值则返回true，否则返回false
     * 
     * \details 带超时的等待版本。如果在指定时间内达到阈值则返回true，
     *          如果超时则返回false。这对于避免无限等待很有用。
     */       
    bool wait(xtime const & timestamp, size_t const task_threshold = 0) const volatile
    {
      const pool_type* self = const_cast<const pool_type*>(this);
      recursive_mutex::scoped_lock lock(self->m_monitor);

      if(0 == task_threshold)
      {
        while(0 != self->m_active_worker_count || !self->m_scheduler.empty())
        { 
          if(!self->m_worker_idle_or_terminated_event.timed_wait(lock, timestamp)) return false;
        }
      }
      else
      {
        while(task_threshold < self->m_active_worker_count + self->m_scheduler.size())
        { 
          if(!self->m_worker_idle_or_terminated_event.timed_wait(lock, timestamp)) return false;
        }
      }

      return true;
    }


  private:	

    /*!
     * \brief 终止所有工作线程
     * \param wait 是否等待所有线程完全终止
     * 
     * \details 触发所有工作线程的终止过程。如果wait为true，
     *          函数会阻塞直到所有线程完全终止并清理资源。
     */
    void terminate_all_workers(bool const wait) volatile
    {
      pool_type* self = const_cast<pool_type*>(this);
      recursive_mutex::scoped_lock lock(self->m_monitor);

      self->m_terminate_all_workers = true;

      m_target_worker_count = 0;
      self->m_task_or_terminate_workers_event.notify_all();

      if(wait)
      {
        while(m_active_worker_count > 0)
        {
          self->m_worker_idle_or_terminated_event.wait(lock);
        }

        for(typename std::vector<shared_ptr<worker_type> >::iterator it = self->m_terminated_workers.begin();
          it != self->m_terminated_workers.end();
          ++it)
        {
          (*it)->join();
        }
        self->m_terminated_workers.clear();
      }
    }


    /*!
     * \brief 更改池中工作线程的数量
     * \param worker_count 新的工作线程数量
     * \return 如果池将被调整大小则返回true，否则返回false
     * 
     * \details 大小调整由SizePolicy处理。函数会根据目标线程数
     *          动态增加或减少工作线程。增加线程时会立即创建，
     *          减少线程时会通知多余的线程终止。
     */
    bool resize(size_t const worker_count) volatile
    {
      locking_ptr<pool_type, recursive_mutex> lockedThis(*this, m_monitor); 

      if(!m_terminate_all_workers)
      {
        m_target_worker_count = worker_count;
      }
      else
      { 
        return false;
      }


      if(m_worker_count <= m_target_worker_count)
      { // 增加工作线程数量
        while(m_worker_count < m_target_worker_count)
        {
          try
          {
            worker_thread<pool_type>::create_and_attach(lockedThis->shared_from_this());
            m_worker_count++;
            m_active_worker_count++;	
          }
          catch(thread_resource_error)
          {
            return false;
          }
        }
      }
      else
      { // 减少工作线程数量
        lockedThis->m_task_or_terminate_workers_event.notify_all();   // TODO: 优化通知的工作线程数量
      }

      return true;
    }


    /*!
     * \brief 工作线程因未处理异常而意外终止
     * \param worker 意外终止的工作线程
     * 
     * \details 处理工作线程异常终止的情况。更新线程计数，
     *          通知等待的线程，并根据是否正在终止所有线程
     *          来决定后续处理。
     */
    void worker_died_unexpectedly(shared_ptr<worker_type> worker) volatile
    {
      locking_ptr<pool_type, recursive_mutex> lockedThis(*this, m_monitor);

      m_worker_count--;
      m_active_worker_count--;
      lockedThis->m_worker_idle_or_terminated_event.notify_all();	

      if(m_terminate_all_workers)
      {
        lockedThis->m_terminated_workers.push_back(worker);
      }
      else
      {
        lockedThis->m_size_policy->worker_died_unexpectedly(m_worker_count);
      }
    }

    /*!
     * \brief 工作线程已析构
     * \param worker 已析构的工作线程
     * 
     * \details 处理工作线程正常析构的情况。更新线程计数，
     *          通知等待的线程，并在终止所有线程时保存线程引用。
     */
    void worker_destructed(shared_ptr<worker_type> worker) volatile
    {
      locking_ptr<pool_type, recursive_mutex> lockedThis(*this, m_monitor);
      m_worker_count--;
      m_active_worker_count--;
      lockedThis->m_worker_idle_or_terminated_event.notify_all();	

      if(m_terminate_all_workers)
      {
        lockedThis->m_terminated_workers.push_back(worker);
      }
    }


    /*!
     * \brief 执行单个任务
     * \return 如果成功执行任务则返回true，如果工作线程应该终止则返回false
     * 
     * \details 工作线程的核心执行循环。从调度器获取任务并执行，
     *          处理线程数量调整，管理线程状态转换。
     *          
     *          执行流程：
     *          1. 检查是否需要减少线程数量
     *          2. 等待任务或终止信号
     *          3. 从调度器获取任务
     *          4. 执行任务函数
     */
    bool execute_task() volatile
    {
      function0<void> task;

      { // 获取任务
        pool_type* lockedThis = const_cast<pool_type*>(this);
        recursive_mutex::scoped_lock lock(lockedThis->m_monitor);

        // 如有必要，减少线程数量
        if(m_worker_count > m_target_worker_count)
        {	
          return false;	// 终止工作线程
        }


        // 等待任务
        while(lockedThis->m_scheduler.empty())
        {	
          // 如有必要，减少工作线程数量
          if(m_worker_count > m_target_worker_count)
          {	
            return false;	// 终止工作线程
          }
          else
          {
            m_active_worker_count--;
            lockedThis->m_worker_idle_or_terminated_event.notify_all();	
            lockedThis->m_task_or_terminate_workers_event.wait(lock);
            m_active_worker_count++;
          }
        }

        task = lockedThis->m_scheduler.top();
        lockedThis->m_scheduler.pop();
      }

      // 调用任务函数
      if(task)
      {
        task();
      }
 
      //guard->disable();
      return true;
    }
  };



} } } // namespace boost::threadpool::detail 

#endif // THREADPOOL_POOL_CORE_HPP_INCLUDED
